package com.ch.storm.demo.bolt;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Tuple;
import com.ch.storm.demo.utils.JRedisUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;

/**
 * @author zhimin
 */
public class WordCounter extends BaseBasicBolt {

    private static final long serialVersionUID = 1L;

    private Logger log = LoggerFactory.getLogger(WordCounter.class);

    Integer id;

    String name;

    Map<String, Integer> counters;

    /**
     * On create  初始化
     */
    @Override
    public void prepare(@SuppressWarnings("rawtypes") Map stormConf, TopologyContext context) {
        this.counters = new HashMap<>();
        this.name = context.getThisComponentId();
        this.id = context.getThisTaskId();
    }

    /**
     * At the end of the spout (when the cluster is shutdown
     * We will show the word counters
     * <p>
     * Topology执行完毕的清理工作，比如关闭连接、释放资源等操作都会写在这里
     * 因为这只是个Demo，我们用它来打印我们的计数器
     */

    @Override
    public void cleanup() {
        log.info("-- Word Counter [{}-{}] --", name, id);
        for (Map.Entry<String, Integer> entry : counters.entrySet()) {
            log.info("map[{}:{}]", entry.getKey(), entry.getValue());
            log.info("redis[{}:{}]", entry.getKey(), JRedisUtil.getJRedis().get(entry.getKey()));
            JRedisUtil.getJRedis().del(entry.getKey());
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    }


    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        String str = input.getString(0);
        /**
         * If the word don't exist in the map we will create
         * this, if not We will add 1
         *
         * 如果单词尚不存在于map，我们就创建一个，如果已在，我们就为它加1
         *
         */

        log.info(">>>>>>>>>>>>统计：{}<<<<<<<<<<<<<<", str);
        if (!counters.containsKey(str)) {
            counters.put(str, 1);
        } else {
            Integer c = counters.get(str) + 1;
            counters.put(str, c);
        }

        JRedisUtil.getJRedis().incr(str);
    }
}
